package com.vodafone.mapreduce.divide;

import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.log4j.Logger;

import com.vodafone.mapreduce.MapReduce;
import com.vodafone.mapreduce.dto.function.MapInput;

/**
 * TODO: Insert your comments here!
 */
public abstract class DivideFunction<K, V> {

    public static Logger log = MapReduce.LOG;

    protected final List<Session> jmsSessions;
    protected final List<MessageProducer> messageProducers;

    protected final K k;
    protected final V v;
    protected final Map<K, List<V>> intermediate = new HashMap<K, List<V>>();
    protected final AtomicBoolean ready = new AtomicBoolean(false);

    public DivideFunction(K k, V v) {
        this.k = k;
        this.v = v;
        this.jmsSessions = null;
        this.messageProducers = null;
    }

    public DivideFunction(K k, V v, List<Session> jmsSessions, List<MessageProducer> messageProducers) {
        this.k = k;
        this.v = v;
        this.jmsSessions = jmsSessions;
        this.messageProducers = messageProducers;
    }

    public void divide() {
        doDivide();
        ready.set(true);
    }

    protected abstract void doDivide();

    protected void sendChunk(K k, V v) {
        if (messageProducers != null) {
            // Send out the crap to JMS.
            try {
                MapInput<K, V> c = new MapInput<K, V>().withKey(k).withValue(v);
                Message m = jmsSessions.get(0).createObjectMessage((Serializable) c);
                m.setStringProperty("created-by", "DivideFunction");
                messageProducers.get(0).send(m);
            } catch (JMSException ex) {
                // Crap!
                throw new RuntimeException("Failed to send out a chunk to JMS!", ex);
            }
        }
    }

    public Map<K, List<V>> getDivideResult() {
        if (ready.get()) {
            return intermediate;
        } else {
            return null;
        }
    }

}
